package dataframe

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._

object DataFrame_FinalDemo01 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local[*]")
    conf.setAppName("DataFrame_FinalDemo01")

    val spark: SparkSession = SparkSession
      .builder()
      .config(conf)
      .getOrCreate()
    import spark.implicits._

    val ordersInfo: DataFrame = spark
      .read
      .option("header", true)
      .option("sep", ",")
      .option("inferSchema", true)
      .csv("data/NW-Orders-Info.csv")
    ordersInfo.printSchema()
    ordersInfo.show(5)

    val ordersDetail: DataFrame = spark
      .read
      .option("header", true)
      .option("sep", ",")
      .option("inferSchema", true)
      .csv("data/NW-Orders-Details.csv")
    ordersDetail.printSchema()
    ordersDetail.show(5)

    // 1、每个客户下了多少订单?
    ordersInfo
      .groupBy("CustomerID")
      .agg(
        count("OrderID").as("OrderCountByCustomerID")
      )
      .sort(desc("OrderCountByCustomerID"))
      .show()

    // 2、每个国家的订单有多少?
    ordersInfo
      .groupBy("ShipCountry")
      .agg(
        count("OrderID").as("OrderCountByShipCountry")
      )
      .sort(desc("OrderCountByShipCountry"))
      .show()

    // - 每月/年有多少订单?
    // - 每个客户的年销售总额是多少?
    // - 客户每年的平均订单是多少?

    // 增加一个date，month和year三个列
    ordersInfo
      .withColumn("Date",to_date($"OrderDate"))
      .withColumn("Year",year($"Date"))
      .withColumn("Month",month($"Date"))
      .groupBy("Year","Month")
      .agg(
        count("OrderID").as("OrderCountByYearAndMonth")
      )
      .select(
      "Year",
      "Month",
        "OrderCountByYearAndMonth"
    )
      .orderBy("Year","Month")
      .show()

    // 每个客户的年销售总额是多少?
    // 1.1. 计算每个订单明细的实际金额
    // 1.2. 根据order id统计每张订单的总金额
    val OrderPrice: DataFrame = ordersDetail
      .select(
        $"OrderID",
        (($"UnitPrice" - $"UnitPrice" * $"Discount") * $"Qty").as("OrderPrice")
      )
      .groupBy("OrderID")
      .agg(
        sum("OrderPrice").as("OrderTotalPrice")
      )
    // OrderPrice.where($"OrderID"===10248).show()
    ordersInfo
      .join(
        OrderPrice,
        ordersInfo("OrderID").equalTo(OrderPrice("OrderID")),
        "inner"
      )
      .select(
        ordersInfo("OrderID"),
        ordersInfo("CustomerID"),
        ordersInfo("OrderDate"),
        ordersInfo("ShipCountry"),
        OrderPrice("OrderTotalPrice")
      )
      .groupBy($"CustomerID",year($"OrderDate").as("year"))
      .agg(
        bround(sum("OrderTotalPrice"),2).as("OrderTotalPriceByCustomerIDAndYear")
      )
      .sort("CustomerID","year")
      .show()


    ordersInfo
      .join(
        OrderPrice,
        ordersInfo("OrderID").equalTo(OrderPrice("OrderID")),
        "inner"
      )
      .select(
        ordersInfo("OrderID"),
        ordersInfo("CustomerID"),
        ordersInfo("OrderDate"),
        ordersInfo("ShipCountry"),
        OrderPrice("OrderTotalPrice")
      )
      .groupBy($"CustomerID",year($"OrderDate").as("year"))
      .agg(
        bround(avg("OrderTotalPrice"),2).as("OrderAvgPriceByCustomerIDAndYear")
      )
      .sort("CustomerID","year")
      .show()


    spark.stop()
  }
}
